package com.spi.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.spi.config.MyWebSocketHandler;
import com.spi.init.ServerInfo;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.Map;

@RocketMQMessageListener(topic = "${rocketmq.consumer.group4.topic}",
        consumerGroup = "${rocketmq.consumer.group4.consumer-group}",
        consumeMode = ConsumeMode.CONCURRENTLY, //并发消费
        //messageModel = MessageModel.BROADCASTING,
        selectorExpression = "${rocketmq.consumer.group4.tag}")
@Component
@Slf4j
public class NotifyWebSocketListener implements RocketMQListener<MessageExt> {

    @Autowired
    private ServerInfo serverInfo ;

    @Override
    public void onMessage(MessageExt messageExt) {
        System.out.println("survey-topic-notifyWebSocket 消费者组即将消费。。");
        log.info("survey-topic-notifyWebSocket 消费者组即将消费。。");
        byte[] bytes = messageExt.getBody();
        String content = new String(bytes, StandardCharsets.UTF_8);
        Map<String, Object> stringObjectMap = JSON.parseObject(content, new TypeReference<Map<String, Object>>() {});
        String sockId = (String)stringObjectMap.get("sockId");
        MyWebSocketHandler.sendMessageToClient(sockId, "后台处理完成111");
    }
}
